package com.jeffse.light.stream.core.function.base;

import com.jeffse.light.stream.core.data.context.KeyedContext;
import com.jeffse.light.stream.core.data.context.ProcessContext;
import com.jeffse.light.stream.core.data.descriptor.stateDescriptor.ListStateDescriptor;
import com.jeffse.light.stream.core.data.descriptor.stateDescriptor.ValueStateDescriptor;
import com.jeffse.light.stream.core.data.state.ListState;
import com.jeffse.light.stream.core.data.state.ValueState;
import com.jeffse.light.stream.core.function.advance.WindowFunction;
import com.jeffse.light.stream.core.pojo.window.WindowInfo;

import java.util.ArrayList;
import java.util.List;

/**
 *
 * @param <KEY>
 */
public class CountWindowKeyedProcessFunction<KEY, T, O>  extends KeyedProcessFunction<KEY, T, O> {
    private WindowInfo windowInfo = null;
    private WindowFunction windowFunction = null;
    private ListState<T> listState;

    public CountWindowKeyedProcessFunction(WindowInfo windowInfo, WindowFunction windowFunction){
        this.windowInfo = windowInfo;
        this.windowFunction = windowFunction;
    }

    @Override
    public void open(){
        super.open();
        listState = getRuntimeContext().getListState(new ListStateDescriptor("listState", this));
    }

    @Override
    public void processElement(ProcessContext context, T value) throws Exception {
        List<T> list = new ArrayList<>();
        if(listState.getList() != null){
            list.addAll(listState.getList());
        }

        list.add(value);
        listState.update(list);
        if(list.size() >= windowInfo.getNum()){
            compute();
        }
    }

    public void compute() throws Exception {
        windowFunction.window(listState.getList(), this);
        listState.clear();
    }
}